package mayor.mq.message.model;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

public class Consumer1 {
    public Consumer1(){
        try {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("message-consumer");


            consumer.setNamesrvAddr("192.168.2.246:9876;192.168.2.190:9876");
            consumer.subscribe("Topic1", "Tag1 || Tag2 || Tag3");
            consumer.setMessageModel(MessageModel.BROADCASTING);

            consumer.registerMessageListener(new Listener());

            consumer.start();
        }catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("Consumer Started. ");
    }
 class Listener implements MessageListenerConcurrently{

        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
           try{
                for(MessageExt msg : msgs) {
                    String topic  = msg.getTopic();
                    String msgBody = new String(msg.getBody(),"utf-8");
                    String tags = msg.getTags();
                    System.out.println("收到消息" + "topic : " + topic +"tags : "+tags+"msg" + msgs+"msgBody"+msgBody);
                    //休息一分钟表示处理失败
                    //System.out.println("开始停");
                    //Thread.sleep(600000);
                }

           }catch (Exception e){
               e.printStackTrace();
               return ConsumeConcurrentlyStatus.RECONSUME_LATER;
           }

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    }

    public static void main(String[] args) {
        Consumer1 consumer1 = new Consumer1();
        System.out.println("c1 start ");
    }
}
